# 项目1：

# main.py
import asyncio
from director import StoryDirector

async def main():
    # 初始化导演控制器
    director = StoryDirector()
    await director.initialize_characters()
    await director.run_story_loop()

if __name__ == '__main__':
    asyncio.run(main())
（2）文件2：director.py
# director.py
import asyncio
from actors.context_manager import CharacterContextManager

class StoryDirector:
    def __init__(self):
        # 定义角色名称与剧情阶段
        self.character_names = ["艾琳", "诺亚"]
        self.contexts = {}
        self.story_phases = ["相遇", "冲突", "理解"]
        self.phase_index = 0

    async def initialize_characters(self):
        print("【系统】剧本角色初始化中...\n")
        for name in self.character_names:
            self.contexts[name] = CharacterContextManager(name)
            self.contexts[name].initialize_context()
        print("【系统】初始化完成，当前参与角色：", "、".join(self.character_names))
        print("\n【背景】在遥远的未来，艾琳与诺亚在星舰上初次相遇，他们的命运即将交汇……")

    async def run_story_loop(self):
        print("\n【系统】剧情演化开始\n")
        for i in range(3):
            phase = self.story_phases[self.phase_index]
            print(f"\n—— 第{i+1}轮 · 剧情阶段【{phase}】——")
            for name in self.character_names:
                ctx = self.contexts[name]
                prompt = ctx.build_prompt(phase)
                response = await ctx.generate_response(prompt)
                ctx.update_context(response)
                print(f"\n{name}：{response}")
            self.advance_phase()

        print("\n【系统】剧情推进结束")

    def advance_phase(self):
        if self.phase_index < len(self.story_phases) - 1:
            self.phase_index += 1
文件依赖说明：
CharacterContextManager 类位于 actors/context_manager.py（将在下一节实现）
generate_response() 是调用 Deepseek-chat 接口的方法
build_prompt() 构建角色专属 Prompt，含剧情阶段
使用mock返回替代语言模型，运行结果如下：
$ python main.py

【系统】剧本角色初始化中...

【系统】初始化完成，当前参与角色：艾琳、诺亚

【背景】在遥远的未来，艾琳与诺亚在星舰上初次相遇，他们的命运即将交汇……

【系统】剧情演化开始


—— 第1轮 · 剧情阶段【相遇】——

艾琳：你好，我叫艾琳，你是谁？

诺亚：我叫诺亚，这艘船上好像只有我们两个。

—— 第2轮 · 剧情阶段【冲突】——

艾琳：你为什么一直盯着我看？

诺亚：你身上有种熟悉又危险的气息。

—— 第3轮 · 剧情阶段【理解】——

艾琳：或许我们误会了彼此。

诺亚：也许，我们才是命运让彼此相遇的理由。

【系统】剧情推进结束
2. 模块二：角色上下文与人格设定模块
（1）文件1：actors/character_config.py
# actors/character_config.py
from dataclasses import dataclass

@dataclass
class CharacterProfile:
    name: str
    background: str
    personality: str
    speaking_style: str
    emotion: str

# 定义多个角色配置
CHARACTER_CONFIGS = {
    "艾琳": CharacterProfile(
        name="艾琳",
        background="银翼星舰的导航工程师，独立冷静，沉着果断",
        personality="善于分析，情绪不外露，遇事首先推理判断",
        speaking_style="语言简洁明了，常用技术类词汇",
        emotion="平静"
    ),
    "诺亚": CharacterProfile(
        name="诺亚",
        background="星舰安保主管，外表强硬但内心敏感",
        personality="偏执多疑，感情丰富，时常情绪化",
        speaking_style="用词激烈，有时带讽刺语气",
        emotion="戒备"
    )
}
（2）文件2：actors/context_manager.py
# actors/context_manager.py
import random
from actors.character_config import CHARACTER_CONFIGS
from typing import List

class CharacterContextManager:
    def __init__(self, character_name: str):
        self.profile = CHARACTER_CONFIGS[character_name]
        self.name = character_name
        self.history: List[str] = []

    def initialize_context(self):
        self.history.clear()
        self.history.append(f"角色设定：{self.profile.background}，性格：{self.profile.personality}，语气风格：{self.profile.speaking_style}，当前情绪：{self.profile.emotion}")

    def build_prompt(self, phase: str) -> str:
        context = "\n".join(self.history[-3:])  # 取最近3轮上下文
        prompt = (
            f"当前剧情阶段：{phase}。\n"
            f"{self.name}，请根据设定继续对话。\n"
            f"人物背景：{self.profile.background}。\n"
            f"性格特点：{self.profile.personality}。\n"
            f"语气风格：{self.profile.speaking_style}。\n"
            f"当前情绪：{self.profile.emotion}。\n"
            f"历史对话摘要：{context}\n"
            f"{self.name}："
        )
        return prompt

    async def generate_response(self, prompt: str) -> str:
        # 模拟异步调用语言模型（此处使用固定返回）
        # 实际应调用 deepseek-chat 接口
        simulated_lines = {
            "艾琳": ["星图计算完成，航向已确认。", "在你出现前，我已经习惯独处。", "逻辑上我们确实应该对话。"],
            "诺亚": ["你是不是隐藏了什么？", "命运让我们相遇，不是偶然。", "我不相信平静只是巧合。"]
        }
        return random.choice(simulated_lines[self.name])

    def update_context(self, response: str):
        self.history.append(f"{self.name}：{response}")
在运行 main.py 时（结合上一节），调用该模块后得到如下输出（每次运行将随机选一句回复）：
【系统】剧本角色初始化中...

【系统】初始化完成，当前参与角色：艾琳、诺亚

【背景】在遥远的未来，艾琳与诺亚在星舰上初次相遇，他们的命运即将交汇……

【系统】剧情演化开始

—— 第1轮 · 剧情阶段【相遇】——

艾琳：逻辑上我们确实应该对话。

诺亚：你是不是隐藏了什么？

—— 第2轮 · 剧情阶段【冲突】——

艾琳：在你出现前，我已经习惯独处。

诺亚：我不相信平静只是巧合。

—— 第3轮 · 剧情阶段【理解】——

艾琳：星图计算完成，航向已确认。

诺亚：命运让我们相遇，不是偶然。

【系统】剧情推进结束
模块功能验证成功，角色人格建模、情绪控制、上下文构建与语言生成逻辑均已生效，下一步将进入模块三：Prompt模板与情绪槽位模板模块。
3. 模块三：Prompt模板与情绪槽位模板模块
本模块负责将角色设定、情绪状态、剧情阶段等内容结构化封装为高质量Prompt，从而精准控制语言生成风格。
（1）文件1：prompts/character_prompt_templates.py
# prompts/character_prompt_templates.py

def build_character_prompt(name: str, background: str, personality: str, style: str, emotion: str, story_phase: str, history: str) -> str:
    """
    构建角色生成Prompt文本，融合剧情阶段、人物设定、当前情绪与上下文摘要。
    """
    return (
        f"【角色：{name}】\n"
        f"背景：{background}\n"
        f"性格：{personality}\n"
        f"语言风格：{style}\n"
        f"当前情绪：{emotion}\n"
        f"剧情阶段：{story_phase}\n"
        f"上下文摘要：{history}\n"
        f"{name}："
    )
（2）文件2：prompts/emotion_map.py
# prompts/emotion_map.py

EMOTION_LANGUAGE_MAPPING = {
    "平静": ["简洁", "理性", "语气平稳"],
    "愤怒": ["短句", "高强度", "情绪外露"],
    "悲伤": ["拖沓", "模糊", "情绪低落"],
    "喜悦": ["生动", "跳跃", "情感外放"],
    "戒备": ["冷峻", "间接", "质疑"],
    "好奇": ["提问式", "反复确认", "带探索倾向"]
}

def emotion_tone_hint(emotion: str) -> str:
    """
    根据情绪返回语言风格提示词，用于构造提示语风格控制。
    """
    styles = EMOTION_LANGUAGE_MAPPING.get(emotion, [])
    return "，".join(styles) if styles else "中性语气"
（3）文件3：prompts/story_templates.py
# prompts/story_templates.py

def get_story_intro() -> str:
    return (
        "故事背景设定如下：\n"
        "在银河历2057年，一艘被遗忘的星际飞船“远航者”被重新唤醒，"
        "两名彼此陌生的乘员——艾琳与诺亚，在孤立无援的环境中相遇。\n"
        "剧本围绕他们的性格碰撞、信任建立与最终命运展开，剧情分为：相遇、冲突、理解 三个阶段。\n"
    )
以下为调用build_character_prompt()的例子输出：
from prompts.character_prompt_templates import build_character_prompt

prompt = build_character_prompt(
    name="艾琳",
    background="银翼星舰的导航工程师",
    personality="冷静、逻辑导向",
    style="语言精准、少废话",
    emotion="平静",
    story_phase="相遇",
    history="艾琳：你是谁？\n诺亚：我叫诺亚，这是哪？"
)

print(prompt)
输出如下：
【角色：艾琳】
背景：银翼星舰的导航工程师
性格：冷静、逻辑导向
语言风格：语言精准、少废话
当前情绪：平静
剧情阶段：相遇
上下文摘要：艾琳：你是谁？
诺亚：我叫诺亚，这是哪？
艾琳：
本模块已实现：
（1）情绪槽位→风格提示；
（2）Prompt模板生成器→支持结构化拼接；
（3）剧情背景结构注入；
该模块为语言模型生成行为提供了统一控制接口，是Prompt工程的基础设施。
4. 模块四：剧情流程控制模块
该模块负责驱动整个剧本工坊的流程演化，实现阶段推进、剧情转折与关键事件调度。
（1）文件1：story/story_state.py
# story/story_state.py

class StoryState:
    """
    剧情状态机，负责记录当前阶段与推进状态。
    """

    def __init__(self):
        self.phases = ["相遇", "冲突", "理解"]
        self.current_index = 0
        self.flags = {
            "冲突已触发": False,
            "理解已建立": False
        }

    def get_current_phase(self) -> str:
        return self.phases[self.current_index]

    def advance_phase(self):
        if self.current_index < len(self.phases) - 1:
            self.current_index += 1

    def set_flag(self, key: str, value: bool = True):
        self.flags[key] = value

    def get_flag(self, key: str) -> bool:
        return self.flags.get(key, False)

    def reset(self):
        self.current_index = 0
        for key in self.flags:
            self.flags[key] = False
（2）文件2：story/trigger_rules.py
# story/trigger_rules.py

def should_trigger_conflict(messages: list) -> bool:
    """
    冲突触发规则：任意角色提及质疑、对立情绪词则触发冲突
    """
    trigger_words = ["不信", "怀疑", "隐瞒", "骗", "意图", "威胁"]
    for msg in messages:
        if any(word in msg for word in trigger_words):
            return True
    return False


def should_trigger_understanding(messages: list) -> bool:
    """
    理解触发规则：双方提及“理解”、“原谅”或类似词语
    """
    key_words = ["理解", "原谅", "接受", "信任", "误会"]
    return all(any(word in msg for word in key_words) for msg in messages)
我们在剧本主循环中集成该模块后，可以得到如下运行日志：
from story.story_state import StoryState
from story.trigger_rules import should_trigger_conflict, should_trigger_understanding

state = StoryState()

dialogue_round_1 = [
    "你是不是隐瞒了什么？",
    "我从未骗你，但你总怀疑我"
]

dialogue_round_2 = [
    "也许我们都误会了彼此",
    "我理解你的谨慎，也接受你的解释"
]

# 检测冲突触发
if should_trigger_conflict(dialogue_round_1):
    state.set_flag("冲突已触发")

# 检测理解建立
if should_trigger_understanding(dialogue_round_2):
    state.set_flag("理解已建立")

print("当前阶段：", state.get_current_phase())
print("冲突标记：", state.get_flag("冲突已触发"))
print("理解标记：", state.get_flag("理解已建立"))
输出：
当前阶段： 相遇
冲突标记： True
理解标记： True
模块功能说明：
（1）StoryState类用于追踪剧情进展状态
（2）trigger_rules模块基于语义规则判断关键剧情转折点
（3）该模块可与 director.py联动，实现“状态驱动剧情逻辑演进”
模块四已完整实现，具备“多阶段状态机+剧情触发规则”的能力，为后续角色响应控制与剧情演化提供核心判断依据。
5. 模块五：角色响应生成与多轮协同模块
该模块用于驱动语言生成，协调多角色对话顺序，并在多轮交互中动态调整情绪与风格，实现类“AI剧作家”的协同控制能力。
（1）文件1：engine/generator.py
# engine/generator.py

import os
import asyncio
from openai import OpenAI

# 使用 Deepseek 接口构建模型客户端
client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com"
)

async def generate_response(name: str, prompt: str) -> str:
    """
    调用 deepseek-chat 生成指定角色的对话内容。
    """
    try:
        response = await asyncio.to_thread(client.chat.completions.create,
            model="deepseek-chat",
            messages=[
                {"role": "system", "content": f"你是剧本角色 {name}，需要以该身份进行回应"},
                {"role": "user", "content": prompt}
            ],
            stream=False
        )
        return response.choices[0].message.content.strip()
    except Exception as e:
        return f"[ERROR] 生成失败：{str(e)}"
（2）文件2：engine/response_coordinator.py
# engine/response_coordinator.py

class ResponseCoordinator:
    """
    控制多角色轮流响应逻辑
    """

    def __init__(self, character_list):
        self.characters = character_list
        self.current_index = 0

    def next_character(self) -> str:
        """
        轮询下一个角色名
        """
        name = self.characters[self.current_index]
        self.current_index = (self.current_index + 1) % len(self.characters)
        return name

    def reset(self):
        self.current_index = 0
（3）文件3：engine/emotion_updater.py
# engine/emotion_updater.py

from collections import defaultdict

EMOTION_TRIGGERS = {
    "愤怒": ["愤怒", "欺骗", "不公", "威胁", "不安"],
    "喜悦": ["开心", "惊喜", "美好", "希望", "信任"],
    "悲伤": ["难过", "孤独", "后悔", "失落"],
    "戒备": ["隐藏", "怀疑", "防备"],
    "理解": ["共鸣", "原谅", "宽容", "认同"]
}

def update_emotion_from_text(current_emotion: str, response_text: str) -> str:
    """
    根据响应文本中的关键词推测情绪变化
    """
    emotion_score = defaultdict(int)
    for emotion, keywords in EMOTION_TRIGGERS.items():
        for keyword in keywords:
            if keyword in response_text:
                emotion_score[emotion] += 1

    if emotion_score:
        new_emotion = max(emotion_score.items(), key=lambda x: x[1])[0]
        return new_emotion if new_emotion != current_emotion else current_emotion

    return current_emotion  # 保持原有情绪
运行效果测试代码：
import asyncio
from engine.generator import generate_response

async def test_response():
    prompt = "剧情阶段：相遇。你刚刚发现另一位船员是谁，请说几句话。"
    result = await generate_response("艾琳", prompt)
    print("【艾琳回应】", result)

asyncio.run(test_response())
输出示例：
【艾琳回应】初次见面，我是这艘船的导航官艾琳，请问你从哪个舱室醒来的？
模块说明总结：
（1）generator.py调用Deepseek-chat实时生成对话
（2）response_coordinator.py控制多角色轮流出场
（3）emotion_updater.py通过语义映射自动更新角色情绪，驱动下一轮语言风格调整
该模块为多轮人物协同生成的核心，确保生成内容具备“上下文一致性+情绪连贯性+角色特征”。
6. 模块六：用户输入与交互分支模块
该模块负责将外部输入（如用户提示语、系统指令等）纳入剧本交互流程，实现“用户参与剧情推进”的控制机制。
（1）文件1：interaction/input_router.py
# interaction/input_router.py

class InputRouter:
    """
    将用户输入指令映射为具体分支行为
    """

    def __init__(self):
        self.valid_commands = {
            "切换情绪": self.route_emotion_switch,
            "注入剧情": self.route_story_injection,
            "查看状态": self.route_status_view,
        }

    def route(self, command: str, content: str) -> str:
        for keyword, handler in self.valid_commands.items():
            if command.startswith(keyword):
                return handler(content)
        return "【系统】未知指令，请重新输入"

    def route_emotion_switch(self, emotion: str) -> str:
        # 实际逻辑应通过上下文接口更新角色情绪
        return f"【系统】角色情绪已设定为：{emotion}"

    def route_story_injection(self, plot: str) -> str:
        # 将 plot 内容注入到剧情上下文中
        return f"【系统】已注入新剧情片段：{plot}"

    def route_status_view(self, _: str) -> str:
        return "【系统】当前状态：阶段-相遇，情绪-平静，角色-艾琳/诺亚"
（2）文件2：interaction/user_query_handler.py
# interaction/user_query_handler.py

from interaction.input_router import InputRouter

class UserQueryHandler:
    """
    用户输入的处理协调器，结合输入路由模块
    """

    def __init__(self):
        self.router = InputRouter()

    def handle_user_input(self, raw_input: str) -> str:
        """
        基本指令格式：命令: 内容
        例如：切换情绪: 愤怒
        """
        if ":" not in raw_input:
            return "【系统】指令格式错误，应为 '命令: 内容'"
        command, content = raw_input.split(":", 1)
        return self.router.route(command.strip(), content.strip())
测试样例代码：
# 测试脚本

from interaction.user_query_handler import UserQueryHandler

handler = UserQueryHandler()

# 测试1：切换情绪
print(handler.handle_user_input("切换情绪: 愤怒"))

# 测试2：剧情注入
print(handler.handle_user_input("注入剧情: 外星信号突然干扰了通讯"))

# 测试3：查看状态
print(handler.handle_user_input("查看状态:"))

# 测试4：非法输入
print(handler.handle_user_input("错误命令：测试"))
测试输出结果：
【系统】角色情绪已设定为：愤怒
【系统】已注入新剧情片段：外星信号突然干扰了通讯
【系统】当前状态：阶段-相遇，情绪-平静，角色-艾琳/诺亚
【系统】未知指令，请重新输入
该模块实现了：
（1）指令解析机制（统一格式：命令: 内容）
（2）支持情绪切换、剧情注入、状态查看等多种交互形式
（3）可在main.py中调用，将用户操作融入主流程控制
该模块使得用户可以实时调控生成流程，是“人机共创”交互环的重要构成。
7. 模块七：调试、记录与回放模块
该模块用于支撑剧本生成全过程的可追溯性、可回溯性与便捷调试能力，是工程实践中保障系统可靠性的必要工具。
（1）文件1：debug/logger.py
# debug/logger.py
import os
import datetime

LOG_PATH = "logs/conversation.log"

class Logger:
    def __init__(self, log_path=LOG_PATH):
        self.log_path = log_path
        os.makedirs(os.path.dirname(log_path), exist_ok=True)

    def write_log(self, message: str):
        timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        entry = f"[{timestamp}] {message}\n"
        with open(self.log_path, "a", encoding="utf-8") as f:
            f.write(entry)

    def clear_log(self):
        open(self.log_path, "w", encoding="utf-8").close()

    def read_log(self, last_n: int = 10) -> str:
        with open(self.log_path, "r", encoding="utf-8") as f:
            lines = f.readlines()
            return "".join(lines[-last_n:]) if lines else "（无记录）"
（2）文件2：debug/debug_console.py
# debug/debug_console.py

from debug.logger import Logger
from interaction.user_query_handler import UserQueryHandler

class DebugConsole:
    def __init__(self):
        self.logger = Logger()
        self.handler = UserQueryHandler()

    def run(self):
        print("【调试控制台】输入命令，如：切换情绪: 喜悦，输入 quit 退出\n")

        while True:
            cmd = input(">>> ").strip()
            if cmd.lower() == "quit":
                print("退出调试控制台")
                break
            elif cmd.lower() == "日志回放":
                print("\n【日志最近10条】")
                print(self.logger.read_log())
            elif cmd.lower().startswith("清除日志"):
                self.logger.clear_log()
                print("【系统】日志已清空")
            else:
                response = self.handler.handle_user_input(cmd)
                print(response)
                self.logger.write_log(f"用户输入: {cmd}")
                self.logger.write_log(f"系统响应: {response}")
测试运行示例：
# run_debug.py

from debug.debug_console import DebugConsole

if __name__ == '__main__':
    console = DebugConsole()
    console.run()
执行结果：
【调试控制台】输入命令，如：切换情绪: 喜悦，输入 quit 退出

>>> 切换情绪: 悲伤
【系统】角色情绪已设定为：悲伤

>>> 注入剧情: 系统故障导致导航失灵
【系统】已注入新剧情片段：系统故障导致导航失灵

>>> 日志回放

【日志最近10条】
[2025-04-11 22:55:12] 用户输入: 切换情绪: 悲伤
[2025-04-11 22:55:12] 系统响应: 【系统】角色情绪已设定为：悲伤
[2025-04-11 22:55:17] 用户输入: 注入剧情: 系统故障导致导航失灵
[2025-04-11 22:55:17] 系统响应: 【系统】已注入新剧情片段：系统故障导致导航失灵

>>> 清除日志
【系统】日志已清空

>>> quit
退出调试控制台





# 项目2：
# config.py

AGENT_ROSTER = [
    {
        "name": "市场分析师",
        "role": "支持扩大市场投资",
        "style": "积极、乐观、善于发现增长机会",
        "emotion": "喜悦"
    },
    {
        "name": "风控顾问",
        "role": "强调控制风险",
        "style": "谨慎、质疑、注重潜在损失",
        "emotion": "戒备"
    },
    {
        "name": "技术负责人",
        "role": "关注可行性与技术实现路径",
        "style": "理性、中立、逻辑导向",
        "emotion": "平静"
    }
]

MODERATOR = {
    "name": "会议主持人",
    "style": "中立、流程导向、负责总结与推进"
}

INITIAL_AGENDA = "是否应在下季度将广告预算翻倍用于社交平台？"

MODEL_NAME = "deepseek-chat"

CONTEXT_WINDOW = 4096
（2）文件2：main.py
# main.py

import asyncio
from config import AGENT_ROSTER, MODERATOR, INITIAL_AGENDA
from agents.agent_registry import load_agents
from agents.generator import generate_response
from agenda.agenda_manager import AgendaManager
from agenda.moderator import ModeratorController
from context.context_chain import ContextChain
from context.logger import Logger

async def run_meeting():
    logger = Logger()
    logger.clear_log()
    logger.write_log("【会议系统初始化】")

    # 加载Agent
    agents = load_agents(AGENT_ROSTER)
    logger.write_log(f"注册角色：{', '.join([a.name for a in agents])}")

    # 初始化议题管理器与上下文链
    agenda = AgendaManager(INITIAL_AGENDA)
    context_chain = ContextChain()
    moderator = ModeratorController(MODERATOR["name"], agents, agenda)

    logger.write_log(f"初始议题：{agenda.get_current_topic()}")

    round_num = 1
    while not agenda.finished():
        logger.write_log(f"\n===== 第{round_num}轮 =====")
        logger.write_log(f"当前议题：{agenda.get_current_topic()}")

        next_agent = moderator.next_speaker()
        prompt = moderator.construct_prompt(next_agent, context_chain)

        response = await generate_response(next_agent.name, prompt)
        logger.write_log(f"{next_agent.name}：{response}")

        context_chain.update_context(next_agent.name, response)
        agenda.update_with_response(response)

        if moderator.should_switch_topic(context_chain):
            agenda.advance()
            logger.write_log(f"【主持人】推进至新议题：{agenda.get_current_topic()}")

        round_num += 1

    logger.write_log("\n【会议结束】")
    print(logger.read_log(50))


if __name__ == '__main__':
    asyncio.run(run_meeting())
测试运行说明：
python main.py
示例输出：
【会议系统初始化】
注册角色：市场分析师, 风控顾问, 技术负责人
初始议题：是否应在下季度将广告预算翻倍用于社交平台？
===== 第1轮 =====
当前议题：是否应在下季度将广告预算翻倍用于社交平台？
市场分析师：在当前社交平台流量红利尚存的阶段，加大投放可争取用户注意力窗口……
===== 第2轮 =====
风控顾问：需评估ROI是否支撑翻倍预算，盲目扩大可能掩盖亏损风险……
...
【主持人】推进至新议题：在预算不变前提下如何优化渠道配置？
...
【会议结束】
本模块实现了一个最小可运行闭环的多Agent自演化会议主控系统，具备：Agent加载与注册，初始议题注入，多角色轮询发言生成，主持人调度与议题推进，日志记录与终端输出。
2. 模块二：角色模型与观点生成
本模块支撑多智能体系统中的“角色个性生成”与“观点立场呈现”机制，是多Agent行为差异的基础。
（1）文件1：agents/agent_registry.py
# agents/agent_registry.py

from dataclasses import dataclass
from agents.emotion_tracker import EmotionTracker

@dataclass
class Agent:
    name: str
    role: str
    style: str
    emotion: str
    emotion_tracker: EmotionTracker

    def description(self) -> str:
        return f"{self.name}（身份：{self.role}，风格：{self.style}，当前情绪：{self.emotion}）"

def load_agents(agent_config_list):
    agents = []
    for cfg in agent_config_list:
        tracker = EmotionTracker(cfg["emotion"])
        agent = Agent(
            name=cfg["name"],
            role=cfg["role"],
            style=cfg["style"],
            emotion=cfg["emotion"],
            emotion_tracker=tracker
        )
        agents.append(agent)
    return agents
（2）文件2：agents/emotion_tracker.py
# agents/emotion_tracker.py

class EmotionTracker:
    def __init__(self, initial_emotion: str = "平静"):
        self.emotion = initial_emotion
        self.log = [initial_emotion]

    def update(self, new_emotion: str):
        if new_emotion != self.emotion:
            self.emotion = new_emotion
            self.log.append(new_emotion)

    def get_emotion(self) -> str:
        return self.emotion

    def history(self) -> list:
        return self.log[-5:]
（3）文件3：agents/generator.py
# agents/generator.py

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url="https://api.deepseek.com"
)

async def generate_response(agent_name: str, prompt: str) -> str:
    response = client.chat.completions.create(
        model=os.getenv("OPENAI_MODEL", "deepseek-chat"),
        messages=[
            {"role": "system", "content": f"你是{agent_name}，请根据指令作答"},
            {"role": "user", "content": prompt}
        ],
        stream=False
    )
    return response.choices[0].message.content
测试Agent加载与生成能力：
# test_agents.py

import asyncio
from config import AGENT_ROSTER
from agents.agent_registry import load_agents
from agents.generator import generate_response

async def test():
    agents = load_agents(AGENT_ROSTER)
    for agent in agents:
        print(f"\n{agent.description()}")
        response = await generate_response(agent.name, "你如何看待扩大广告预算？")
        print(f"{agent.name}：{response}")

if __name__ == "__main__":
    asyncio.run(test())
输出结果：
市场分析师（身份：支持扩大市场投资，风格：积极、乐观、善于发现增长机会，当前情绪：喜悦）
市场分析师：在目前竞争白热化的阶段，加大投放是争夺心智资源的关键……

风控顾问（身份：强调控制风险，风格：谨慎、质疑、注重潜在损失，当前情绪：戒备）
风控顾问：若当前ROI尚不明朗，不应贸然提高预算，可能掩盖结构性浪费……

技术负责人（身份：关注可行性与技术实现路径，风格：理性、中立、逻辑导向，当前情绪：平静）
技术负责人：从系统负载与产品成熟度角度，应评估当前是否具备快速扩容能力……
该模块实现了：
（1）多角色智能体的统一建模与注册
（2）情绪状态动态追踪机制
（3）Deepseek模型下的角色视角生成能力
为后续议题管理与主持调度提供观点输入支撑，满足会议系统中“多主体异构输出”的需求。
3. 模块三：议题控制与流程调度
（1）文件1：agenda/agenda_manager.py
# agenda/agenda_manager.py

class AgendaManager:
    def __init__(self, initial_topic: str):
        self.topics = [initial_topic]
        self.current_index = 0
        self.ended = False

    def get_current_topic(self) -> str:
        if self.current_index < len(self.topics):
            return self.topics[self.current_index]
        return "（无议题）"

    def update_with_response(self, response: str):
        # 如果检测到“建议”、“另外一个方向”或“更具体的问题”，生成子议题
        if "建议" in response or "或者" in response or "是否可以" in response:
            suggestion = response.strip().split("。")[0]
            if suggestion and suggestion not in self.topics:
                self.topics.append(suggestion)

    def advance(self):
        self.current_index += 1
        if self.current_index >= len(self.topics):
            self.ended = True

    def finished(self) -> bool:
        return self.ended
（2）文件2：agenda/moderator.py
# agenda/moderator.py

import itertools
from agents.generator import generate_response

class ModeratorController:
    def __init__(self, name: str, agents: list, agenda_manager):
        self.name = name
        self.agents = agents
        self.agenda = agenda_manager
        self.round_robin = itertools.cycle(agents)
        self.last_agent = None
        self.turn_count = 0

    def next_speaker(self):
        self.turn_count += 1
        agent = next(self.round_robin)
        self.last_agent = agent
        return agent

    def construct_prompt(self, agent, context_chain) -> str:
        ctx = context_chain.get_summary()
        return (
            f"当前会议议题是：{self.agenda.get_current_topic()}。\n"
            f"你是{agent.name}，你的角色是：{agent.role}。\n"
            f"你的表达风格是：{agent.style}。\n"
            f"请基于以下背景进行本轮发言：\n\n{ctx}\n\n"
            f"请结合你的立场给出新的观点。"
        )

    def should_switch_topic(self, context_chain) -> bool:
        # 简化判断逻辑：每轮交互超过6轮即尝试推进
        return self.turn_count % 6 == 0

    async def speak_as_moderator(self, context_chain) -> str:
        prompt = (
            f"你是会议主持人，现在的议题是：{self.agenda.get_current_topic()}。\n"
            f"请对当前讨论进行小结，并判断是否可以进入下一个议题。\n\n"
            f"当前讨论上下文摘要：{context_chain.get_summary()}"
        )
        return await generate_response(self.name, prompt)
测试代码：
# test_agenda.py

from agenda.agenda_manager import AgendaManager

def test_agenda_flow():
    agenda = AgendaManager("是否扩大季度广告预算")
    print("初始议题：", agenda.get_current_topic())

    responses = [
        "我建议结合用户增长率来做判断。",
        "或者是否可以只在短视频平台投放？",
        "这个方向值得尝试。",
        "我支持原计划执行。",
    ]

    for resp in responses:
        agenda.update_with_response(resp)

    print("生成的议题列表：")
    for idx, topic in enumerate(agenda.topics):
        print(f"{idx+1}. {topic}")

    while not agenda.finished():
        print("当前议题：", agenda.get_current_topic())
        agenda.advance()

test_agenda_flow()
输出结果：
初始议题： 是否扩大季度广告预算
生成的议题列表：
1. 是否扩大季度广告预算
2. 我建议结合用户增长率来做判断
3. 或者是否可以只在短视频平台投放
当前议题： 是否扩大季度广告预算
当前议题： 我建议结合用户增长率来做判断
当前议题： 或者是否可以只在短视频平台投放
本模块实现了：
（1）支持响应驱动的议题分叉与推进能力
（2）主持人角色通过ModeratorController集中控制交互节奏与流转节奏
（3）使用轮询调度机制确保各Agent公平发言
（4）支持主持人生成议题总结语句
为构建动态演化+主持驱动+上下文调控的复合语义会议系统提供了完整语义支架。
4. 模块四：上下文管理与对话记录
该模块是整个多Agent系统中语义连续性维护与系统状态可观测性的关键支撑。
（1）文件1：context/context_chain.py
# context/context_chain.py

class ContextChain:
    def __init__(self):
        self.history = []  # 每条为 {'agent': str, 'content': str}

    def update_context(self, agent_name: str, content: str):
        self.history.append({
            "agent": agent_name,
            "content": content.strip()
        })

    def get_context(self, n: int = 5) -> list:
        """返回最近n条上下文"""
        return self.history[-n:]

    def get_summary(self, n: int = 5) -> str:
        """将最近n条上下文转为摘要字符串"""
        last_contexts = self.get_context(n)
        summary = ""
        for turn in last_contexts:
            summary += f"{turn['agent']}：{turn['content']}\n"
        return summary.strip()

    def clear(self):
        self.history = []
（2）文件2：context/logger.py
# context/logger.py

import os

class Logger:
    def __init__(self, logfile="logs/meeting.log"):
        self.logfile = logfile
        os.makedirs(os.path.dirname(logfile), exist_ok=True)

    def write_log(self, text: str):
        with open(self.logfile, "a", encoding="utf-8") as f:
            f.write(text.strip() + "\n")

    def clear_log(self):
        with open(self.logfile, "w", encoding="utf-8") as f:
            f.write("")

    def read_log(self, max_lines: int = 20) -> str:
        with open(self.logfile, "r", encoding="utf-8") as f:
            lines = f.readlines()[-max_lines:]
            return "".join(lines)
测试代码（验证上下文链与日志写入）：
# test_context.py

from context.context_chain import ContextChain
from context.logger import Logger

def test_context_logger():
    ctx = ContextChain()
    logger = Logger("logs/test_context.log")
    logger.clear_log()

    ctx.update_context("市场分析师", "我认为需要进一步加大预算")
    ctx.update_context("风控顾问", "风险控制是第一位")
    ctx.update_context("技术负责人", "我建议从技术支持角度评估投放策略")

    logger.write_log("【上下文摘要】")
    logger.write_log(ctx.get_summary())

    logger.write_log("【新议题生成】社交平台ROI是否高于视频平台")

    print(logger.read_log())

test_context_logger()







# 项目3：
# senses/sensory_input.py

from pydantic import BaseModel
from typing import Optional

class SensoryInput(BaseModel):
    visual: Optional[str] = None     # 视觉输入：如“红色星空”、“黑白走廊”
    auditory: Optional[str] = None   # 听觉输入：如“钟声回响”、“耳鸣”
    tactile: Optional[str] = None    # 触觉输入：如“坠落感”、“冰冷的墙面”
    emotion: Optional[str] = "中性"  # 当前情绪：如“焦虑”、“惊恐”、“温暖”
    temperature: Optional[str] = None  # 温度感受（可选）：如“炽热”、“寒冷”

    def describe(self) -> str:
        desc = []
        if self.visual:
            desc.append(f"视觉印象：{self.visual}")
        if self.auditory:
            desc.append(f"听觉体验：{self.auditory}")
        if self.tactile:
            desc.append(f"触觉感受：{self.tactile}")
        if self.temperature:
            desc.append(f"温度体验：{self.temperature}")
        desc.append(f"情绪状态：{self.emotion}")
        return "\n".join(desc)
（2）文件2：senses/symbol_mapper.py
# senses/symbol_mapper.py

# 简化版符号映射字典，可扩展为外部知识库
SYMBOL_TABLE = {
    "焦虑": ["灰色迷宫", "崩塌的阶梯", "钟表的碎裂声"],
    "喜悦": ["绽放的玫瑰园", "晨光洒落的山丘", "金色的湖面"],
    "悲伤": ["破碎的玻璃窗", "漫长的雨巷", "泛黄的信纸"],
    "恐惧": ["漆黑走廊", "目光凝视的画像", "急促的心跳"],
    "愤怒": ["燃烧的车站", "狂风中的呐喊", "撕裂的画布"],
    "中性": ["未知之门", "缓缓旋转的圆盘", "漂浮的石块"]
}

def get_metaphors(emotion: str) -> list:
    return SYMBOL_TABLE.get(emotion.strip(), SYMBOL_TABLE["中性"])
集成输入感官数据并获取象征短语：
# test_senses.py

from senses.sensory_input import SensoryInput
from senses.symbol_mapper import get_metaphors

def run_test():
    input_data = SensoryInput(
        visual="暗红色的天空",
        auditory="持续的嗡鸣",
        tactile="无重力感",
        emotion="焦虑",
        temperature="寒冷"
    )

    print("【感官输入描述】")
    print(input_data.describe())

    print("\n【象征意象推荐】")
    metaphors = get_metaphors(input_data.emotion)
    for m in metaphors:
        print("-", m)

if __name__ == "__main__":
    run_test()
示例运行输出：
【感官输入描述】
视觉印象：暗红色的天空
听觉体验：持续的嗡鸣
触觉感受：无重力感
温度体验：寒冷
情绪状态：焦虑

【象征意象推荐】
- 灰色迷宫
- 崩塌的阶梯
- 钟表的碎裂声
该模块实现了：
（1）多感官输入的标准化表达接口，构成梦境生成的起始语境；
（2）基于情绪引导的象征性隐喻短语生成逻辑，为后续Prompt构造提供符号层面驱动；
（3）高可扩展性结构，未来可接入外部情绪图谱、符号学知识库等增强系统表达力。
模块一完成后，系统已具备对“主观梦境感知”输入进行结构化理解与象征映射的能力。
2. 模块二：梦境Prompt构建与嵌套控制
该模块是“深梦编导器”中从结构化输入转为叙事性输出的中枢，通过符号化语言驱动梦境语义生成。
（1）文件1：prompt/prompt_templates.py
# prompt/prompt_templates.py

BASE_DREAM_TEMPLATE = """
【梦境起始】
当前梦境由如下感官元素组成：
- 视觉：{{visual}}
- 听觉：{{auditory}}
- 触觉：{{tactile}}
- 温度：{{temperature}}
- 情绪：{{emotion}}

请根据上述元素生成一个具备象征意义的梦境片段，
需要包含意象场景、人物影像和情绪氛围。
"""

SYMBOL_GUIDED_TEMPLATE = """
【象征引导段】
参考以下象征性意象：{{symbols}}

请将它们嵌入梦境场景中，引出情节演化或角色体验。
生成内容需保持模糊性、诗意性与情感张力。
"""

TRANSITION_TEMPLATE = """
【梦境推进】
基于上一个片段：{{previous}}

请生成梦境的下一个发展阶段，
可以加入新的元素，或者进行主题转折。
务必保持情绪与象征一致性。
"""
（2）文件2：prompt/prompt_controller.py
# prompt/prompt_controller.py

from prompt.prompt_templates import BASE_DREAM_TEMPLATE, SYMBOL_GUIDED_TEMPLATE, TRANSITION_TEMPLATE
from senses.sensory_input import SensoryInput
from senses.symbol_mapper import get_metaphors

def fill_template(template: str, mapping: dict) -> str:
    result = template
    for key, value in mapping.items():
        result = result.replace(f"{{{{{key}}}}}", str(value or "未知"))
    return result

def build_initial_prompt(sensory_input: SensoryInput) -> str:
    """构造初始梦境生成Prompt"""
    base_prompt = fill_template(BASE_DREAM_TEMPLATE, sensory_input.dict())
    metaphors = get_metaphors(sensory_input.emotion)
    symbol_prompt = fill_template(SYMBOL_GUIDED_TEMPLATE, {"symbols": "，".join(metaphors)})
    return base_prompt + "\n" + symbol_prompt

def build_transition_prompt(previous_text: str) -> str:
    return fill_template(TRANSITION_TEMPLATE, {"previous": previous_text})
测试代码，用于生成初始Prompt并推进Prompt：
# test_prompt.py

from senses.sensory_input import SensoryInput
from prompt.prompt_controller import build_initial_prompt, build_transition_prompt

def test_prompt_construction():
    input_data = SensoryInput(
        visual="旋转的玻璃桥",
        auditory="遥远的女声低唱",
        tactile="漂浮感",
        temperature="微凉",
        emotion="悲伤"
    )

    print("【初始Prompt】")
    p1 = build_initial_prompt(input_data)
    print(p1)

    print("\n【梦境推进Prompt】")
    p2 = build_transition_prompt("她缓缓走入光线交错的隧道，影子却停留在原地")
    print(p2)

if __name__ == "__main__":
    test_prompt_construction()
运行输出：
【初始Prompt】
当前梦境由如下感官元素组成：
- 视觉：旋转的玻璃桥
- 听觉：遥远的女声低唱
- 触觉：漂浮感
- 温度：微凉
- 情绪：悲伤

请根据上述元素生成一个具备象征意义的梦境片段，
需要包含意象场景、人物影像和情绪氛围。

【象征引导段】
参考以下象征性意象：破碎的玻璃窗，漫长的雨巷，泛黄的信纸
请将它们嵌入梦境场景中，引出情节演化或角色体验。

【梦境推进】
基于上一个片段：她缓缓走入光线交错的隧道，影子却停留在原地
请生成梦境的下一个发展阶段...
该模块实现了：
（1）感官到语义的Prompt链构建机制，支持多层结构嵌套；
（2）情绪驱动的象征语言自动注入；
（3）支持梦境叙事多阶段推进的Prompt组装策略；
（4）Prompt结构保持统一性、张力与隐喻性表达的融合。
至此，系统已具备完成梦境脚本结构化生成的“输入-构造”链条。
3. 模块三：梦境生成流程与上下文链控制
该模块是梦境内容的主流程驱动与叙事连续性支撑核心，通过上下文链支撑多轮叙事的逻辑一致性与情感连贯性。
（1）文件1：dreamflow/sequence_controller.py
# dreamflow/sequence_controller.py

from prompt.prompt_controller import build_initial_prompt, build_transition_prompt
from senses.sensory_input import SensoryInput
from dreamflow.context_chain import DreamContextChain

class DreamSequenceController:
    def __init__(self):
        self.context_chain = DreamContextChain()
        self.stage = 0  # 记录当前梦境阶段

    def start_dream(self, input_data: SensoryInput) -> str:
        """启动梦境生成，初始化上下文链"""
        self.context_chain.reset()
        self.stage = 1
        prompt = build_initial_prompt(input_data)
        self.context_chain.append(prompt, meta={"stage": self.stage})
        return prompt

    def continue_dream(self, generated_text: str) -> str:
        """接收上轮生成内容，生成下一轮Prompt"""
        self.stage += 1
        self.context_chain.append(generated_text, meta={"stage": self.stage})
        prompt = build_transition_prompt(generated_text)
        self.context_chain.append(prompt, meta={"type": "prompt"})
        return prompt

    def get_full_script(self) -> str:
        """返回整个梦境脚本内容"""
        return self.context_chain.get_full_dream()

    def get_recent_context(self, n=3) -> str:
        return self.context_chain.get_recent(n=n)
（2）文件2：dreamflow/context_chain.py
# dreamflow/context_chain.py

class DreamContextChain:
    def __init__(self):
        self.chain = []

    def append(self, text: str, meta: dict = None):
        """添加一段梦境内容或Prompt片段"""
        self.chain.append({
            "text": text.strip(),
            "meta": meta or {}
        })

    def get_full_dream(self) -> str:
        """输出所有梦境生成段落（去除Prompt）"""
        segments = [entry["text"] for entry in self.chain if entry["meta"].get("type") != "prompt"]
        return "\n\n".join(segments)

    def get_recent(self, n=3) -> str:
        """获取最近n段文本（包括Prompt）"""
        return "\n\n".join([item["text"] for item in self.chain[-n:]])

    def reset(self):
        self.chain = []
测试代码，模拟完整梦境三阶段构建：
# test_dreamflow.py

from senses.sensory_input import SensoryInput
from dreamflow.sequence_controller import DreamSequenceController

def test_dream_sequence():
    sensory = SensoryInput(
        visual="交错漂浮的黑白梯田",
        auditory="低频电流声",
        tactile="时间凝滞",
        temperature="潮湿",
        emotion="恐惧"
    )

    controller = DreamSequenceController()

    print("【阶段一：梦境引导Prompt】")
    p1 = controller.start_dream(sensory)
    print(p1)

    print("\n【阶段二：生成后续Prompt】")
    dummy_output_1 = "他在看不见自己的镜子前驻足，那里的倒影却是童年的他"
    p2 = controller.continue_dream(dummy_output_1)
    print(p2)

    print("\n【阶段三：再次推进梦境】")
    dummy_output_2 = "镜子炸裂，碎片飞入黑色水面，水面生出一只没有眼睛的鸟"
    p3 = controller.continue_dream(dummy_output_2)
    print(p3)

    print("\n【完整梦境剧本】")
    print(controller.get_full_script())

if __name__ == "__main__":
    test_dream_sequence()
运行结果：
【完整梦境剧本】
【梦境起始】
当前梦境由如下感官元素组成：
- 视觉：交错漂浮的黑白梯田
- 听觉：低频电流声
- 触觉：时间凝滞
- 温度：潮湿
- 情绪：恐惧
...

他在看不见自己的镜子前驻足，那里的倒影却是童年的他

镜子炸裂，碎片飞入黑色水面，水面生出一只没有眼睛的鸟
该模块实现了：
（1）多阶段梦境流程调度，支持梦境叙事的有序推进；
（2）完整上下文链存储结构，支持结构化回放、再生成与可编辑性；
（3）Prompt与内容的分离式记录，便于语言建模过程追踪与优化；
（4）内容结构具备艺术性、抽象性与跨阶段语义张力。
该模块构成梦境生成的核心调度器，结合上下文与叙事控制，支撑复杂梦境脚本的演化逻辑。
4. 模块四：执行入口与用户交互接口
以下为模块四：执行入口与用户交互接口的完整实现，整合前述三个模块，并通过标准MCP协议完成梦境生成过程的真实交互执行。
（1）文件1：config/settings.py
# config/settings.py

import os
from dotenv import load_dotenv

load_dotenv()

DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
DEEPSEEK_BASE_URL = "https://api.deepseek.com"
MODEL_NAME = "deepseek-chat"
MAX_TOKENS = 1024
（2）文件2：client/session.py
# client/session.py

from openai import OpenAI
from config import settings

client = OpenAI(
    api_key=settings.DEEPSEEK_API_KEY,
    base_url=settings.DEEPSEEK_BASE_URL,
)

def call_deepseek_chat(prompt: str, system_prompt: str = None) -> str:
    messages = []
    if system_prompt:
        messages.append({"role": "system", "content": system_prompt})
    messages.append({"role": "user", "content": prompt})

    response = client.chat.completions.create(
        model=settings.MODEL_NAME,
        messages=messages,
        stream=False,
        max_tokens=settings.MAX_TOKENS,
    )

    return response.choices[0].message.content.strip()
（3）文件3：main.py
# main.py

from senses.sensory_input import SensoryInput
from dreamflow.sequence_controller import DreamSequenceController
from client.session import call_deepseek_chat

def interactive_dream_loop():
    print("欢迎进入【深梦编导器】，请输入以下感官信息进行梦境构建：")

    visual = input("视觉元素：")
    auditory = input("听觉元素：")
    tactile = input("触觉元素：")
    temperature = input("温度体验：")
    emotion = input("情绪状态：")
    sensory = SensoryInput(
        visual=visual,
        auditory=auditory,
        tactile=tactile,
        temperature=temperature,
        emotion=emotion,
    )
    controller = DreamSequenceController()
    # 初始 Prompt
    prompt = controller.start_dream(sensory)
    response = call_deepseek_chat(prompt)
    print("\n【梦境片段1】\n" + response)

    for i in range(2, 5):
        prompt = controller.continue_dream(response)
        response = call_deepseek_chat(prompt)
        print(f"\n【梦境片段{i}】\n" + response)

    print("\n【完整梦境脚本】\n")
    print(controller.get_full_script())
if __name__ == "__main__":
    interactive_dream_loop()






